package org.zjt.flink.wordcount;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.FilterOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.zjt.flink.wordcount.util.WordCountData;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Random;

/**
 * Description:  累计器
 *
 *      ExecutionEnvironment  BATCH  批处理
 *
 *      StreamExecutionEnvironment   STREAM 流式处理
 *
 * @author juntao.zhang
 * Date: 2018-09-29 下午2:40
 * @see
 */
public class StopWordCountAccumulator {

    private static final String EMPTY_COUNT_KEY = "empty_count_key" ;

    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        FilterOperator<Tuple2<String, Integer>> filter = env.fromElements(WordCountData.WORDS).flatMap(new LineSplitter()).filter(new StopWordCountFilter());
        int i = new Random().nextInt(100);

        if (i % 2 == 0){
            filter.writeAsCsv("/Users/juntao.zhang/dev/workspace/myproject/flink-demo/src/main/java/org/zjt/flink/wordcount/resource/");
            JobExecutionResult execute = env.execute();
            Integer accumulatorResult = execute.getAccumulatorResult(EMPTY_COUNT_KEY);
            System.out.println(accumulatorResult);
        }else {
            filter.print();
            Object accumulatorResult1 = env.getLastJobExecutionResult().getAccumulatorResult(EMPTY_COUNT_KEY);
            System.out.println(accumulatorResult1);
        }
    }


    public static final class StopWordCountFilter extends RichFilterFunction<Tuple2<String, Integer>> {

        StopWordAccumulator stopWordAccumulator = new StopWordAccumulator(0);
        @Override
        public boolean filter(Tuple2<String, Integer> tuple2) throws Exception {
            String f0 = tuple2.f0;
            if (Objects.isNull(f0) || Objects.equals(f0,"is") || Objects.equals(f0,"are") || Objects.equals(f0,"was") || Objects.equals(f0,"that")){
                stopWordAccumulator.add(tuple2.f1);
                return false;
            }
            return true;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            // filter的启动时候增加累加器
            getRuntimeContext().addAccumulator(EMPTY_COUNT_KEY,stopWordAccumulator);
        }
    }


    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {

            String[] tokens = s.toLowerCase().split("\\W+");

            for (String token: tokens) {
                if (token.length() > 0) {
                    collector.collect(new Tuple2(token, 1));
                }
            }
        }
    }


    public static class StopWordAccumulator implements Accumulator<Integer,Integer>{

        private Integer accumulator = new Integer(0);

        public StopWordAccumulator(Integer accumulator) {
            this.accumulator = accumulator;
        }

        /**
         * Increases the result vector component at the specified position by 1.
         */
        @Override
        public void add(Integer cnt) {
            accumulator += cnt;
        }



        @Override
        public Integer getLocalValue() {
            return this.accumulator;
        }

        @Override
        public void resetLocal() {
            // clear the result vector if the accumulator instance shall be reused
            accumulator = 0;
        }

        @Override
        public void merge(final Accumulator<Integer, Integer> other) {
            Integer localValue = other.getLocalValue();
            accumulator += localValue;
        }

        @Override
        public Accumulator<Integer, Integer> clone() {
            return new StopWordAccumulator(this.accumulator);
        }

    }
}
